package com.zhang.gmall.realtime.app.dws;

import com.zhang.gmall.realtime.app.func.KeywordUDTF;
import com.zhang.gmall.realtime.beans.KeywordStats;
import com.zhang.gmall.realtime.common.GmallConstant;
import com.zhang.gmall.realtime.utils.ClickHouseUtil;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: 关键词统计
 * @author: zhang
 * @date: 2022/3/16 19:23
 */
public class KeywordStatsSqlApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.获取表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 3.检查点相关设置
        //TODO 4.读取kafka数据流转换为动态表
        String pageTopic = "dwd_page_log_2022";
        String groupId = "KeywordStatsSqlApp";

        tableEnv.executeSql(
                "create table page_log (" +
                        " common MAP<STRING,STRING>," +
                        " page MAP<STRING,STRING>," +
                        " ts BIGINT," +
                        " rt as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000))," +
                        " WATERMARK FOR rt AS rt - INTERVAL '5' SECOND" +
                        ") WITH (" + MyKafkaUtil.getKafkaDDl(pageTopic, groupId) + ")"
        );

        //TODO 5.对读取数据内容过滤出搜索行为。
        Table keywordTable = tableEnv.sqlQuery(
                "select page['item'] keyword," +
                        "rt " +
                        " from page_log " +
                        " where page['last_page_id']='search' " +
                        " and page['item'] is not null "
        );

        //TODO 6.注册UDTF函数
        tableEnv.createTemporaryFunction("ik_analyze", KeywordUDTF.class);
        //TODO 7.对过滤内容进行分词
        Table wordTable = tableEnv.sqlQuery(
                "select word," +
                        "rt " +
                        "from " + keywordTable + "," +
                        "LATERAL TABLE(ik_analyze(keyword)) AS t(word)"
        );

        //TODO 8.分组开窗、聚合
        Table table = tableEnv.sqlQuery(
                " select " +
                        " DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') stt," +
                        " DATE_FORMAT(TUMBLE_END(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') edt," +
                        " word keyword," +
                        " count(*) ct ," +
                        "'" + GmallConstant.KEYWORD_SEARCH + "'source ," +
                        " UNIX_TIMESTAMP()*1000 ts" +
                        " from " + wordTable +
                        " GROUP BY TUMBLE(rt, INTERVAL '10' second), word "
        );

//insert into keyword_stats(keyword,ct,source,stt,edt,ts)  " +
//                        " values(?,?,?,?,?,?)
        //TODO 9.写入clickhouse
        //打印测试
        DataStream<KeywordStats> keywordStatsDS = tableEnv.toAppendStream(table, KeywordStats.class);
        keywordStatsDS.print(">>>>>>");
        keywordStatsDS
                .addSink(ClickHouseUtil.getSinkFunction(
                        "insert into keyword_stats_2022 (keyword,ct,source,stt,edt,ts) " +
                                "values(?,?,?,?,?,?)"
                ));
        //TODO 10.执行任务
        env.execute();
    }
}
